#pragma once
#include "pch.h"
// #include <librdkafka/rdkafkacpp.h>
#include <functional>
#include <librdkafka/rdkafka.h>
#include <stdarg.h>
#include <string>

class KafkaClient {
public:
    SINGLETON_DEFINE(KafkaClient);

    void Init(const std::string &uri, const std::string &topic);

    bool Start();

    bool Stop();

    bool Publish(const std::string &message);

private:
    KafkaClient() = default;

    enum class TYPE {
        PRODUCER = 0,
        CONSUMER,
    };

    enum class STATE {
        IDLE = 0,
        PRE_PRODUCE,
        PRE_CONSUME,
        PRODUCE,
        CONSUME,
    };

    static void logger(const rd_kafka_t *rk, int level, const char *fac, const char *buf);
    static void msg_delivered(rd_kafka_t *rk, const rd_kafka_message_t *msg, void *opaque);
    bool msg_consume(rd_kafka_message_t *msg, uint8_t **p_payload, size_t *p_len);

    TYPE type_ = TYPE::PRODUCER;
    std::string brokers_;
    std::string topic_;
    int32_t partition_ = 0;
    STATE state_ = STATE::IDLE;

    rd_kafka_t *rk_ = nullptr;
    rd_kafka_topic_t *rkt_ = nullptr;
    rd_kafka_conf_t *conf_ = nullptr;
    rd_kafka_topic_conf_t *topic_conf_ = nullptr;
    rd_kafka_message_t *message_ = nullptr;
};